package org.hzandyy.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;

import java.util.UUID;

@SpringBootApplication
@EnableScheduling
@Slf4j
public class KafkaApplication {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    // ******************** 生产者 ********************

    @Scheduled(fixedDelay = 5 * 1000)
    public void producer() {
        String message = System.currentTimeMillis() + "_" + UUID.randomUUID().toString();
        log.info("send [my-topic] message: ({})", message);
        kafkaTemplate.send("my-topic", message);
    }

    // ******************** 消费者 ********************

    @KafkaListener(groupId = "my-consumer-group", topics = "my-topic")
    public void consumer1(String message) {
        log.info("consumer 1 === {}", message);
    }

    @KafkaListener(groupId = "my-consumer-group", topics = "my-topic")
    public void consumer2(String message) {
        log.info("consumer 2 === {}", message);
    }

    public static void main(String[] args) {
        SpringApplication.run(KafkaApplication.class, args);
    }

}
